package com.atguigu.day04;

import com.atguigu.bean.OrderEvent;
import com.atguigu.bean.TxEvent;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.ConnectedStreams;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.KeyedCoProcessFunction;
import org.apache.flink.util.Collector;

import java.util.HashMap;

public class Flink03_Project_OrderWatch {
    /**
     * 1.订单表 的数据来了之后先去存放 交易表 的HashMap中查看是否有能关联上的数据
     *                   |
     *                   |
     *                   |->有关联上的， 那么关联成功 ，然后将关联上的数据从对方的集合中删除                   |
     *                   |
     *                   |->没关联上 ，将自己存入 存放订单表的集合中 等待交易表的数据关联
     *
     *
     * 2.交易表 的数据来了之后先去存放 订单表 的HashMap中查看是否有能关联上的数据
     *      *                   |
     *      *                   |
     *      *                   |->有关联上的， 那么关联成功 ，然后将关联上的数据从对方的集合中删除                  |
     *      *                   |
     *      *                   |->没关联上 ，将自己存入 存放交易表的集合中 等待订单表的数据关联
     */
    public static void main(String[] args) throws Exception {
        //1.获取流的执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        //2.分别获取和订单表的数据交易表的数据
        SingleOutputStreamOperator<OrderEvent> orderStream = env.readTextFile("input/OrderLog.csv")
                .map(new MapFunction<String, OrderEvent>() {
                    @Override
                    public OrderEvent map(String value) throws Exception {
                        String[] split = value.split(",");
                        return new OrderEvent(Long.parseLong(split[0]), split[1], split[2], Long.parseLong(split[3]));
                    }
                });

        SingleOutputStreamOperator<TxEvent> txStream = env.readTextFile("input/ReceiptLog.csv")
                .map(new MapFunction<String, TxEvent>() {
                    @Override
                    public TxEvent map(String value) throws Exception {
                        String[] split = value.split(",");
                        return new TxEvent(split[0], split[1], Long.parseLong(split[2]));
                    }
                });

        //3.连接两条流
        ConnectedStreams<OrderEvent, TxEvent> connectedStreams = orderStream.connect(txStream);

        //4.多并行度的情况下一定要做keyby
        ConnectedStreams<OrderEvent, TxEvent> keyByStream = connectedStreams.keyBy("txId", "txId");

        //5.关联两条流的数据
        keyByStream.process(new KeyedCoProcessFunction<String, OrderEvent, TxEvent, String>() {

            //创建一个Map集合用来存放订单表的数据
            private HashMap<String, OrderEvent> orderMap = new HashMap<>();
            //创建一个Map集合用来存放交易表的数据
            private HashMap<String, TxEvent> txMap = new HashMap<>();

            @Override
            public void processElement1(OrderEvent value, Context ctx, Collector<String> out) throws Exception {
                //1.查询交易表的Map集合看是否有能关联上的数据
                if (txMap.containsKey(value.getTxId())) {
                    //有能够关联上的数据
                    out.collect("订单：" + value.getOrderId() + "对账成功！！！！");
                    //删除已关联上的数据 为了防止内存溢出
                    txMap.remove(value.getTxId());
                } else {
                    //没有能够关联上的数据
                    orderMap.put(value.getTxId(), value);
                }
            }

            @Override
            public void processElement2(TxEvent value, Context ctx, Collector<String> out) throws Exception {
                //1.查询订单表的Map集合看是否有能关联上的数据
                if (orderMap.containsKey(value.getTxId())) {
                    //有能够关联上的数据
                    out.collect("订单：" + orderMap.get(value.getTxId()).getOrderId() + "对账成功！！！！");
                    //删除已关联上的数据 为了防止内存溢出
                    orderMap.remove(value.getTxId());
                } else {
                    //没有能够关联上的数据
                    txMap.put(value.getTxId(), value);
                }
            }
        }).print();

        env.execute();

    }
}
